#pragma once

#include <fmtlog/fmtlog.h>

#include <list>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

#include "Message.pb.h"

namespace MY_MQ {
using ComsumerCallback =
    std::function<void(const std::string &, const MQ_Protobuf::BasicProperties *bp, const std::string &)>;

struct Consumer {
    std::string      _ComsumerID;
    std::string      _SubSriber_QueueName;
    bool             _AutoAck;
    ComsumerCallback _call;
    using self_ptr = std::shared_ptr<Consumer>;

    Consumer() {
        logd("Comsumer Init ,Ptr:{}", (void *)this);
    }

    Consumer(const std::string &ComsumerID, const std::string &SubscribeQueueName, bool AutoAck,
             const ComsumerCallback &cb)
        : _ComsumerID(ComsumerID), _SubSriber_QueueName(SubscribeQueueName), _AutoAck(AutoAck), _call(cb) {
        logd("Comsumer Init ,Ptr:{}", (void *)this);
    }

    ~Consumer() {
        logd("Comsumer Destroy ,Ptr:{}", (void *)this);
    }
};

class ConsumerQueue {

public:
    using self_ptr = std::shared_ptr<ConsumerQueue>;

    ConsumerQueue(const std::string &name) : _name(name) {
    }

    Consumer::self_ptr Create(const std::string &ComsumerID, const std::string &SubscribeQueueName, bool AutoAck,
                              const ComsumerCallback &cb) {
        std::lock_guard<std::mutex> lock(_mutx);
        for (auto &c : _qc) {
            if (c->_ComsumerID == ComsumerID) {
                logw("ComsumerID:{} already exist", ComsumerID);
                return c;
            }
        }

        auto c = std::make_shared<Consumer>(ComsumerID, SubscribeQueueName, AutoAck, cb);
        _qc.push_back(c);
        return c;
    }

    void Remove(const std::string &ComsumerID) {
        std::lock_guard<std::mutex> lock(_mutx);
        for (auto it = _qc.begin(); it != _qc.end(); ++it) {
            if ((*it)->_ComsumerID == ComsumerID) {
                _qc.erase(it);
                return;
            }
        }
    }

    Consumer::self_ptr ChooseByTurn() {
        std::lock_guard<std::mutex> lock(_mutx);
        // 选择当前消费者后通过 splice移动到最后
        for (auto it = _qc.begin(); it != _qc.end(); ++it) {
            auto c = *it;
            _qc.splice(_qc.end(), _qc, it);
            return c;
        }
        return nullptr;
    }

    void clear() {
        std::lock_guard<std::mutex> lock(_mutx);
        _qc.clear();
    }

    bool empty()

    {
        std::lock_guard<std::mutex> lock(_mutx);
        return _qc.empty();
    }

    bool exists(const std::string ConsumerID) {
        std::lock_guard<std::mutex> lock(_mutx);
        for (auto &c : _qc) {
            if (c->_ComsumerID == ConsumerID) {
                return true;
            }
        }
        return false;
    }

    std::list<Consumer::self_ptr> Get() {
        return _qc;
    }

private:
    std::string                   _name;
    std::mutex                    _mutx;
    std::list<Consumer::self_ptr> _qc;
};

class ConsumerManager {
public:
    using self_ptr = std::shared_ptr<ConsumerManager>;

    ConsumerManager() {
    }

    void InitComsumerQueue(const std::string QueueName) {
        std::lock_guard lock(_mutx);
        auto            it = _QueueComsumers.find(QueueName);
        if (it != _QueueComsumers.end()) {
            return;
        }
        auto qcomsumers            = std::make_shared<ConsumerQueue>(QueueName);
        _QueueComsumers[QueueName] = qcomsumers;
    }

    void RemoveComsumerQueue(const std::string QueueName) {
        std::lock_guard lock(_mutx);

        _QueueComsumers.erase(QueueName);
    }

    void removeConsumerFromQueue(std::string ComsumersQueueName, const std::string ConsumerID) {
        ConsumerQueue::self_ptr queue_ptr = nullptr;
        {
            std::lock_guard lock(_mutx);
            auto            q_ptr = _QueueComsumers.find(ComsumersQueueName);
            if (q_ptr == _QueueComsumers.end()) {
                q_ptr->second->Remove(ConsumerID);
                logd("No consumerQueue name as {}", ComsumersQueueName);
                return;
            }
            queue_ptr = q_ptr->second;
        }
        queue_ptr->Remove(ConsumerID);
    }

    Consumer::self_ptr Create(const std::string &ConsumerID, const std::string &SubscribeQueueName, bool AutoAck,
                              const ComsumerCallback &cb) {
        ConsumerQueue::self_ptr qcp;
        {
            std::lock_guard lock(_mutx);
            auto            q_ptr = _QueueComsumers.find(SubscribeQueueName);
            if (q_ptr == _QueueComsumers.end()) {
                logd("No consumerQueue name as {}", SubscribeQueueName);
                return nullptr;
            }
            qcp = q_ptr->second;
        }
        return qcp->Create(ConsumerID, SubscribeQueueName, AutoAck, cb);
    }

    Consumer::self_ptr ChooseByTurn(const std::string &SubscribeQueueName) {
        ConsumerQueue::self_ptr qcp;
        {
            std::lock_guard lock(_mutx);
            auto            q_ptr = _QueueComsumers.find(SubscribeQueueName);
            if (q_ptr == _QueueComsumers.end()) {
                logd("No consumerQueue name as {}", SubscribeQueueName);
                return nullptr;
            }
            qcp = q_ptr->second;
        }
        return qcp->ChooseByTurn();
    }

    bool empty(const std::string QueueName) {
        std::lock_guard lock(_mutx);
        auto            q_ptr = _QueueComsumers.find(QueueName);
        if (q_ptr == _QueueComsumers.end()) {
            logd("No consumerQueue name as {}", QueueName);
            return true;
        }
        return q_ptr->second->empty();
    }

    bool exists(const std::string &ComsumerID, const std::string &QueueName) {
        std::lock_guard lock(_mutx);
        auto            q_ptr = _QueueComsumers.find(QueueName);
        if (q_ptr == _QueueComsumers.end()) {
            logd("No consumerQueue name as {}", QueueName);
            return false;
        }
        return q_ptr->second->exists(ComsumerID);
    }

private:
    std::mutex                                               _mutx;
    std::unordered_map<std::string, ConsumerQueue::self_ptr> _QueueComsumers;
};
};  // namespace MY_MQ